package com.bbmall.ads;

import com.bbmall.SparkSessionUtil;
import org.apache.spark.sql.SparkSession;


/**
 * 2. 会员分析 ADS 表抽取
 */
public class DwsToAdsMemberAnalysis {
    public static void main(String[] args) {
        // 解析参数
        if (args.length < 2) {
            System.err.println("参数错误！用法：");
            System.err.println("全量处理：<is_full=true> <date_range=start_date,end_date>");
            System.err.println("增量处理：<is_full=false> <biz_date=yyyy-MM-dd>");
            System.exit(1);
        }
        boolean isFull = Boolean.parseBoolean(args[0]);
        String dateParam = args[1];

        // 动态生成日期条件
        String whereClause;
        if (isFull) {
            String[] dates = dateParam.split(",");
            whereClause = "dt BETWEEN '" + dates[0] + "' AND '" + dates[1] + "'";
        } else {
            whereClause = "dt = '" + dateParam + "'";
        }

        // 动态生成SQL
        String sql = (isFull ? "INSERT OVERWRITE" : "INSERT INTO") + " TABLE ads.ads_member_analysis PARTITION (dt) " +
                "SELECT " +
                "  SUM(total_member_count) AS total_member_count, " +  // 会员总数
                "  SUM(active_member_count) AS active_member_count, " +  // 活跃会员数
                "  AVG(avg_consume_amount) AS avg_consume_amount, " +  // 平均消费
                "  AVG(repurchase_rate) AS repurchase_rate, " +  // 复购率
                // 会员等级分布：JSON格式{"等级":数量}
                "  to_json(collect_list(map(member_level, total_member_count))) AS member_level_distribution, " +
                // 月度消费趋势：JSON数组[{"month":月份,"amount":金额}]
                "  to_json(collect_list(map(concat(stat_year, '-', stat_month), member_sales_amount))) AS monthly_consume_trend, " +
                // 高价值会员Top5：JSON数组[{"member_id":ID,"total_consume":金额}]
                "  to_json(collect_list(map('会员等级', member_level, '累计消费', member_sales_amount))) AS high_value_members, " +
                // 新会员转化率：JSON格式{"阶段":转化率}
                "  to_json(map('首购率', 0.8, '二次购率', 0.6)) AS new_member_conversion, " +  // 示例值，实际从DWS计算
                // 会员偏好：JSON格式{"分类":占比}
                "  to_json(map('食品', 40, '日用品', 30, '生鲜', 20)) AS member_preference, " +  // 示例值，实际从DWS关联商品表
                "  CURRENT_TIMESTAMP() AS update_time, " +
                "  dt " +
                "FROM dws.dws_member_day " +
                "WHERE " + whereClause + " " +
                "GROUP BY dt, stat_year, stat_month, member_level;";

        // 执行
        SparkSession spark = SparkSessionUtil.getSparkSession("DWS_TO_ADS_MEMBER_ANALYSIS");
        try {
            spark.sql(sql);
            System.out.println("会员分析ADS表" + (isFull ? "全量" : "增量") + "处理完成！");
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(1);
        } finally {
            SparkSessionUtil.close();
        }
    }
}